package com.okex.api.client.impl;

import java.io.Closeable;
import java.io.IOException;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.okex.api.client.BinanceApiCallback;
import com.okex.api.client.OkexApiWebSocketClient;
import com.okex.api.client.constant.OkexApiConstants;
import com.okex.api.client.domain.general.DepthLimit;
import com.okex.api.client.domain.market.OkexDepth;
import com.okex.api.client.domain.market.OkexTicker;

import okhttp3.Dispatcher;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;
import okio.ByteString;

/**
 * Binance API WebSocket client implementation using OkHttp.
 */
public class OkexApiWebSocketClientImpl implements OkexApiWebSocketClient, Closeable {

	private OkHttpClient client;

	public OkexApiWebSocketClientImpl() {
		Dispatcher d = new Dispatcher();
		d.setMaxRequestsPerHost(100);
		this.client = new OkHttpClient.Builder().dispatcher(d).build();
	}

	@Override
	public Closeable onOkSubSpotTicker(String symbol, BinanceApiCallback<OkexTicker[]> callback) {
		final String channel = "{'event':'addChannel','channel':'ok_sub_spot_" + symbol + "_ticker" + "','binary':'0'}";
		return createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, OkexTicker[].class));
	}

	@Override
	public Closeable onOkSubSpotDepth(String symbol, DepthLimit depthLimit, BinanceApiCallback<OkexDepth[]> callback) {
		// TODO Auto-generated method stub
		final String channel = "{'event':'addChannel','channel':'ok_sub_spot_" + symbol + "_depth"
				+ depthLimit.toString() + "','binary':'0'}";
		return createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, OkexDepth[].class));
	}

	@Override
	public void close() {
		client.dispatcher().executorService().shutdown();
	}

	private Closeable createNewWebSocket(String channel, BinanceApiWebSocketListener<?> listener) {
		Request request = new Request.Builder().url(OkexApiConstants.WS_API_BASE_URL).build();
		final WebSocket webSocket = client.newWebSocket(request, listener);
		webSocket.send(channel);
		return () -> {
			final int code = 1000;
			listener.onClosing(webSocket, code, null);
			webSocket.close(code, null);
			listener.onClosed(webSocket, code, null);
		};
	}

	public static void main(String[] a) {
		Dispatcher d = new Dispatcher();
		d.setMaxRequestsPerHost(100);
		OkHttpClient client = new OkHttpClient.Builder().dispatcher(d).build();

		String channel = "{'event':'addChannel','channel':'ok_sub_spot_eth_btc_ticker','binary':0}";

		final String OWS_API_BASE_URL = "wss://real.okex.com:10441/websocket";

		Request request = new Request.Builder().url(OWS_API_BASE_URL).build();
		final WebSocket webSocket = client.newWebSocket(request, new WebSocketListener() {
			@Override
			public void onOpen(WebSocket webSocket, Response response) {
				System.out.println("onOpen: " + response);
				// webSocket.send(channel);
				// webSocket.close(1000, "再见");
			}

			@Override
			public void onMessage(WebSocket webSocket, String text) {
				System.out.println("onMessage: " + text);
				ObjectMapper mapper = new ObjectMapper();
				try {
					OkexTicker[] okexTicker = mapper.readValue(text, OkexTicker[].class);
					System.out.println("onMessage okexTicker: " + okexTicker.length);
				} catch (JsonParseException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (JsonMappingException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

			@Override
			public void onMessage(WebSocket webSocket, ByteString bytes) {
				System.out.print("onMessage byteString: " + bytes);
			}

			@Override
			public void onFailure(WebSocket webSocket, Throwable t, Response response) {
				System.out.print("onFailure response: " + response + "]");
				t.printStackTrace();
			}

		});
		webSocket.send(channel);
		client.dispatcher().executorService().shutdown();
	}

}
